
/**
 * @ProjectName: 
 * @Copyright: 2014 lisheng  All Right Reserved.
 * @address: toughheart@163.com
 * @date: 2015年1月13日 下午12:05:13
 * @Description: 本内容未经本人允许禁止使用、转发.
 */
 
package com.ls.fw.search.lucene.server;

import java.io.File;
import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.MultiReader;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.TopDocs;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ls.fw.data.exception.DataException;
import com.ls.fw.search.lucene.bean.LastPageInfo;
import com.ls.fw.search.lucene.bean.SearchCondition;
import com.ls.fw.search.lucene.bean.Session;
import com.ls.fw.search.lucene.mapper.DocMapper;
import com.ls.fw.search.lucene.mapper.LoadMapper;
import com.ls.fw.search.lucene.util.DateUtil;
import com.ls.fw.search.lucene.util.LuceneUtil;
import com.ls.fw.search.lucene.util.SynchronizdDateUtil;


/**
 * lucene服务处理器默认实现，按照时间设置索引目录
 * @author lisheng
 * @date 2015年1月13日 下午12:05:13
 * @version V1.0 
 */
public class DefaultMutilServer extends AbstractMutilServer<DateTime>{

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private static final Map<String, LuceneServer> serverRuter = new ConcurrentHashMap<String, LuceneServer>();
 
	private String baseDir = System.getProperty("user.dir");
	
	private CountDownLatch latch = null;
	
	//是否加载完成
	private AtomicBoolean loadComplete = new AtomicBoolean(false);
	
	/**
	 * <查询条件hashcode,分页Session>
	 */
	private static final Map<String, Session> context = new ConcurrentHashMap<String, Session>();

//	public ReadWriteLock lock =  null;
	
	final ExecutorService exec = Executors.newFixedThreadPool(50);
	
	/**
	 * 是否为测试环境
	 */
	public static boolean test = false;

	
	public DefaultMutilServer() {
		super();
	}

	
	public DefaultMutilServer(String baseDir) {
		super();
		this.baseDir = baseDir;
	}

	
	public void load(){
		this.load(null);
	}
	
	
	public synchronized void load(final LoadMapper mapper){
		File file = new File(baseDir);
		if (file.exists()) {
			final AtomicInteger loadSize = new AtomicInteger();
			final File[] f = file.listFiles();
			final int size = f.length;
			latch = new CountDownLatch(size);
			loadComplete.set(false);
			for (File file2 : f) {
				final String key = file2.getName();
				exec.execute(new Runnable() {
					
					@Override
					public void run() {
						logger.info("正在加载" + key + "....");
						try {
							createByKey(key);
							if(mapper != null){
								mapper.mapper(key);
							}
						} catch (Exception e) {
							e.printStackTrace();
						} finally{
							latch.countDown();
						}
						if(loadSize.addAndGet(1) >= size){
							logger.info("所有服务加载完成!");
						}else{
							logger.info(key+"加载完成!"+"还剩"+(size - loadSize.get())+"个索引目录！");
						}
					}
				});
			}
			this.canHandl();
		} else {
			loadComplete.set(true);
			this.logger.warn(baseDir + "下不存在索引目录!");
		}
	}
	
	
	/**
	 * 是否能处理
	 * 
	 * @author lisheng
	 * @date 2015年1月18日 下午2:07:35
	 * @version V1.0
	 * @return
	 */
	@Override
	public boolean canHandl(){
		try {
			if(!loadComplete.get() && latch != null){
				this.logger.warn("正在等待初始化完成...!");
				long start = System.currentTimeMillis();
			    latch.await();
			    logger.info("等待初始化完耗时:"+(System.currentTimeMillis()-start));
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally{
			loadComplete.set(true);
		}
		return loadComplete.get();
	}
	
	 
	@Override
	public LuceneServer create(DateTime flag) {
		String key = this.getKey(flag);
		return this.createByKey(key);
	}

	
	@Override
	public LuceneServer createByKey(String key) {
		LuceneServer server = serverRuter.get(key);
		if (server == null) {
			synchronized (serverRuter) {
				server = serverRuter.get(key);
				if (server == null) {
					String dir = this.getDir(key);
					server = new DefaultLuceneServer(dir);
					if (!server.isOpened()) {
						server.open();
					}
					serverRuter.put(key, server);
				}
				if (!server.isOpened()) {
					server.open();
				}
			}
		}
		return server;
	}

	
	@Override
	public LuceneServer get(DateTime flag) {
		return this.getByKey(this.getKey(flag));
	}
	
	
	@Override
	public LuceneServer getByKey(String key){
		LuceneServer s = null;
//		synchronized (serverRuter) {
			s = serverRuter.get(key);
//		}
		return s;
	}
	
	@Override
	public LuceneServer remove(String key){
		LuceneServer s = null;
		synchronized (serverRuter) {
			s = serverRuter.remove(key);
		}
		return s;
	}
	

	@Override
	public List<LuceneServer> get(DateTime start, DateTime end) {
		return this.get(start, end, true);
	}
	
	
	@Override
	public List<LuceneServer> get(DateTime start, DateTime end, boolean desc){
		int size = this.getStep(start, end);
		List<LuceneServer> searcherlist = new ArrayList<LuceneServer>(size);
		for (int i = 0; i < size; i++) {
			String key = "";
			if(desc){
				key = this.getNextKey(end, i);
			}else{
				key = this.getLastKey(start, i);
			}
			LuceneServer server = this.getByKey(key);
			if (server == null) {
				File f = new File(this.getDir(key));
				if (f.exists()) {
					server = this.createByKey(key);
				}
			}
			if (server != null) {
				searcherlist.add(server);
			}
		}
		return searcherlist;
	}
	
	@Override
	public int getStep(DateTime start, DateTime end){
		return DateUtil.getDay(start.toDate(), end.toDate())+ 1;
	}
	

	@Override
	public String getKey(DateTime flag) {
		return flag.toString("yyyyMMdd");
	}
	

	@Override
	public String getNextKey(DateTime start) {
		return this.getNextKey(start, 1);
	}
	
	@Override
	public String getNextKey(DateTime start,int step) {
		return this.getKey(start.plusDays(-step));
	}
	
	@Override
	public String getLastKey(DateTime end) {
		return this.getLastKey(end, 1);
	}
	
	
	@Override
	public String getLastKey(DateTime end,int step) {
		return this.getKey(end.plusDays(step));
	}

	@Override
	public LuceneServer getNextServer(DateTime start) {
		return this.getNextServer(start, 1);
	}

	@Override
	public LuceneServer getLastServer(DateTime end) {
		return this.getLastServer(end, 1);
	}

	@Override
	public LuceneServer getNextServer(DateTime start, int step) {
		return this.get(start.plusDays(-step));
	}

	@Override
	public LuceneServer getLastServer(DateTime end, int step) {
		return this.get(end.plusDays(step));
	}

	@Override
	public String getBaseDir() {
		return this.baseDir;
	}

	@Override
	public String getDir(String key) {
		String dir = this.getBaseDir() + File.separator + key;
		return dir;
	}

	@Override
	public IndexSearcher getIndexSearcher(DateTime start, DateTime end) {
		IndexSearcher searcher = null;
		List<LuceneServer> serverlist = this.get(start, end);
		if (!serverlist.isEmpty() && serverlist.size() > 0) {
			if (serverlist.size() > 1) {
				List<IndexReader> searcherlist = this.getIndexReaders(serverlist);
				try {
					searcher = new IndexSearcher(new MultiReader(
							searcherlist.toArray(new IndexReader[searcherlist
									.size()]), false));
				} catch (IOException e) {
					logger.error(e.getMessage(), e);
				}
			} else {
				searcher = serverlist.get(0).acquire();
			}
		}
		return searcher;
	}

	
	@Override
	public List<IndexReader> getIndexReaders(DateTime start, DateTime end) {
		return this.getIndexReaders(this.get(start, end));
	}

	@Override
	public void commit(String... keys) {
		if(keys==null){
			for (String name : serverRuter.keySet()) {
				LuceneServer s = serverRuter.get(name);
				if(s != null){
					s.commit();
					s.refushTimestamp();
				}
			}
		}else{
			for (String name : keys) {
				LuceneServer s = serverRuter.get(name);
				if(s != null){
					s.commit();
					s.refushTimestamp();
				}
			}
		}
	}

	@Override
	public void close(String... keys) {
		synchronized (serverRuter) {
			if(keys == null){
				for (String name : serverRuter.keySet()) {
					LuceneServer s = serverRuter.get(name);
					if(s != null){
						try {
							s.close();
						} catch (IOException e) {
							e.printStackTrace();
						}
						s = null;
					}
				}
				serverRuter.clear();
			}else{
				for (String name : keys) {
					LuceneServer s = serverRuter.get(name);
					if(s != null){
						try {
							s.close();
						} catch (IOException e) {
							e.printStackTrace();
						}
						s = null;
					}
				}
			}
		}
	}
	
	@Override
	public <M> List<M> scan(SearchCondition searchCondition, DateTime start, DateTime end, DocMapper<M> mapper){
		final int step = this.getStep(start, end);
		return this.scan(searchCondition, end, step, mapper);
	}
	
	@Override
	public <M> List<M> scan(SearchCondition searchCondition, DateTime end, int step, final DocMapper<M> mapper){
		if(step <= 0){
			return null;
		}
		long start = System.currentTimeMillis();
		final List<M> result = new Vector<M>();
		int  handlServerSize = 0;//已经处理的服务数量
		final ExecutorService exec = Executors.newFixedThreadPool(step);
		final CountDownLatch latch = new CountDownLatch(step);
		final SearchCondition sc = (SearchCondition) searchCondition.clone();
		while(handlServerSize < step){
			final LuceneServer server = this.getNextServer(end, handlServerSize++);
			if(server == null){
				latch.countDown();
				continue;
			}else{
				exec.execute(new Runnable() {
					@Override
					public void run() {
							try {
								List<M> list = null;
								list = server.scan(sc, mapper);
								if(list != null){
									result.addAll(list);
								}
							} finally{
								latch.countDown();
							}
					}
				});
			}
		}
		exec.shutdown();
		try {
			latch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		this.logger.info("scan耗时："+(System.currentTimeMillis()-start));
		return result;
	}
	
	
	public <M> List<M> scanBlocking(SearchCondition searchCondition, DateTime end, int step, DocMapper<M> mapper){
		long start = System.currentTimeMillis();
		final List<M> result = new ArrayList<M>();
		int  handlServerSize = 0;//已经处理的服务数量
		int total = 0, handCount = 0; 
		IndexSearcher searcher = null;
		boolean nextServer = true;
		
		final SearchCondition sc = (SearchCondition) searchCondition.clone();
		while(handlServerSize < step){
			LuceneServer server = this.getNextServer(end, handlServerSize);
			if(server == null){
				handlServerSize++;
				continue;
			}
			searcher = server.acquire();
			if (searcher == null) {
				handlServerSize++;
				continue;
			}
			try {
				this.logger.info("查询库："+server.getAddress());
				TopDocs td = this.search(searcher, sc);
				if(td != null){
					total = td.totalHits;
					int find = td.scoreDocs.length;
					this.logger.info("总数："+total+",本次找到"+find);
					if(find > 0){
						nextServer = false;
						sc.setLastPageLastDoc(td.scoreDocs[find - 1]);
						List<M> list = LuceneUtil.getResult(0, sc.getPageSize(), searcher
								, td, sc.getShowFields(), mapper);
						if(list != null && !list.isEmpty()){
							result.addAll(list);
							nextServer = false;
						}else{
							nextServer = true;
						}
						handCount += find;
						this.logger.info("总数："+total+",已经找到"+handCount);
						if(handCount >= total){
							nextServer = true;
						}
					} else{
						nextServer = true;
					}
				}else{
					nextServer = true;
				}
			}catch (DataException e) {
				nextServer = true;
				throw e;
			}catch (Exception e) {
				nextServer = true;
				throw new DataException(e.getMessage(),e);
			}finally{
				if(nextServer){
					handlServerSize++;
					sc.setLastPageLastDoc(null);
					total = 0;
					handCount = 0;
					if(server != null){
						server.release(searcher);
					}
				}
			}
		}
		this.logger.info("scan耗时："+(System.currentTimeMillis()-start));
		return result;
	}
	
	/**
	 * 获取索引目录名称
	 * @param dir
	 * @return
	 */
	protected String getName(final String dir){
		File file = new File(dir) ;
		return file.getName();
	}
	
	
	@Override
	public <T> List<T> search(SearchCondition searchCondition, DateTime start, DateTime _end, final DocMapper<T> mapper){
		String code = searchCondition.hashCodeStr();
		Session session = context.get(code);
		LastPageInfo info = null;
		DateTime end = _end;
		if(session != null){ 
			 info = session.getPageInfo(searchCondition.getPage()-1);
			 if(info != null){
					try {
						Date edate = SynchronizdDateUtil.parseDate(
								SynchronizdDateUtil.formatDate(
										SynchronizdDateUtil.parseyyyymmddDate(this.getName(info.getDir()))));
						end = new DateTime(edate.getTime());
					} catch (ParseException e) {
						e.printStackTrace();
					}
				 }
		}
		List<T> result = null;
		int step = this.getStep(start, end);
		if(step > 1){
			result = this.searchFromMutilServer(searchCondition, end, session, step, mapper);
		}else{
			LuceneServer server = this.getNextServer(end, 0);
			if(server != null){
				this.logger.info("查询库："+server.getAddress());
				result = this.searchFromSingleServer(searchCondition, session, step, server, mapper);
			}
		}
		return result;
	}
	 
	
	protected <T> List<T> searchFromMutilServer(SearchCondition searchCondition, DateTime end
			, Session session, int size
			, DocMapper<T> mapper) {
		long start = System.currentTimeMillis();
		List<T> result = new ArrayList<T>();
		int count = 0//已经查询到的数据量
				, handlServerSize = 0;//已经处理的服务数量
		int n = 0, p = searchCondition.getPage(); 
		IndexSearcher searcher = null;
		final int page = searchCondition.getPage(), pageSize = searchCondition.getPageSize();
		while(handlServerSize < size){
			n = searchCondition.getPageSize() - count;
			LuceneServer server = this.getNextServer(end, handlServerSize);
			handlServerSize++;
			if(server == null){
				continue;
			}
			this.logger.info("查询库："+server.getAddress());
			searcher = server.acquire();
			if (searcher == null) {
				continue;
			}
			List<T> list;
			int offset = 0;//开始位置
			if(session != null){
				offset = session.getUsedSizeFromDir(page-1, server.getAddress());
			}
			try {
				list = this.search(server.getAddress()
						, searcher, searchCondition, session, offset, p,  n, mapper);
			} catch (Exception e) {
				throw new IllegalArgumentException(e);
			}finally{
				server.release(searcher);
			}
			if(list != null){
				count += list.size();
				result.addAll(list);
			}
			p = 1;//下一个服务从第一页开始查询
			if(count >= pageSize){
				break;
			}
		}
		this.logger.info("searchFromMutilServer耗时："+(System.currentTimeMillis()-start));
		return result;
	}
	
	
	private int getMaxDoc(final IndexSearcher searcher, final ScoreDoc lastBottom
			, final int offset
			, final int page, final int pageSize){
		int maxDoc = LuceneUtil.getMinResultSizeByPage(page, pageSize, searcher);
		maxDoc = Math.max(offset + pageSize, maxDoc);
		maxDoc = Math.min(searcher.getIndexReader().maxDoc(),maxDoc);
		if(lastBottom != null && page > 1){
			maxDoc = pageSize;
		}
		return maxDoc;
	}
	
	protected <T> List<T> search(final String dir, final IndexSearcher searcher
			, final SearchCondition searchCondition,Session session, final int _offset
			, final int currentPage
			, final int currentPageSize, final DocMapper<T> mapper) {
		final int page = searchCondition.getPage();
		if(test){
			this.logger.info("session：" + session);
			this.logger.info("searchCondition：" + searchCondition);
			this.logger.info("currentPage：" + currentPage);
			this.logger.info("currentPageSize：" + currentPageSize);
		}
		String code = searchCondition.hashCodeStr();
		int find = 0;
		List<T> list = null;
		try {
//			ScoreDoc lastDoc = searchCondition.getLastPageLastDoc();
			ScoreDoc lastDoc = (session!=null?session.getDoc(page-1):null);
			int thisPage = currentPage;
			int offset = _offset;
			if(lastDoc != null){
				offset = 0;
				thisPage++;
			}
			final SearchCondition sc = (SearchCondition) searchCondition.clone();
			sc.setPage(thisPage);
			sc.setPageSize(currentPageSize);
			sc.setOffset(offset);
			sc.setLastPageLastDoc(lastDoc);
			sc.setMaxDoc(this.getMaxDoc(searcher, lastDoc, offset, thisPage, currentPageSize));
			TopDocs td = this.search(searcher, sc);
			find = td.scoreDocs.length;
			if(test){
				this.logger.info("totalHits：" + td.totalHits);
				this.logger.info("返回：" + find);
			}
			if(lastDoc == null ){
				if(currentPage>1){
					System.err.println("出现特殊情况！");
//					offset += (currentPage - 1) * currentPageSize;
				}
			}
			list = LuceneUtil.getResult(offset, currentPageSize, searcher, td
					, sc.getShowFields(), mapper);
			if(find>0 && !list.isEmpty()){//查询到结果
				if(session == null){
					session = new Session();
					session.setConditionHashCode(code);
				}
				LastPageInfo info = null;
				if(!(page>1 && lastDoc == null)){//使用游标
					info = session.getPageInfo(page);
					if(info==null){
						info = new LastPageInfo();
						session.put(page, info);
					}
					//注释掉用不使用游标
					info.setLastDoc(td.scoreDocs[find - 1]);
					info.setPage(page);
					info.setCurrentDirPage(currentPage);
					info.setDir(dir);
					info.setCurrentDirUsedSize(find);
					context.put(code, session);
					this.logger.info("添加使用游标分页"+page+":"+code);
				}else{
					info = session.getPageInfo(page);
					if(info==null){
						info = new LastPageInfo();
						session.put(page, info);
					}
					info.setPage(page);
					info.setCurrentDirPage(currentPage);
					info.setDir(dir);
					info.setCurrentDirUsedSize(find);
					context.put(code, session);
					this.logger.info("添加未使用游标分页"+page+":"+code);
				}
			}else{
//				context.remove(code);
//				this.logger.info("删除分页"+code);
			}
		}catch (Exception e) {
			throw new IllegalArgumentException(e);
		}
		return list;
	}
	
	
	protected <T> List<T> searchFromSingleServer(SearchCondition searchCondition, Session session, int size
			,LuceneServer server,DocMapper<T> mapper){
		final int page = searchCondition.getPage(), pageSize = searchCondition.getPageSize();
		
		IndexSearcher searcher = null;
		searcher = server.acquire();
		List<T> result = new ArrayList<T>();
		if (searcher == null) {
			return null;
		}
		int lastPage = 0;
		int offset = 0;
		if(session != null){
			lastPage = session.getLastPageFromDir(page-1, server.getAddress());
			offset = session.getUsedSizeFromDir(page-1, server.getAddress());
		}
		int cpage = page - lastPage;
		List<T> list = null;
		try {
			list = this.search(server.getAddress(), searcher
				, searchCondition, session,  offset
				,cpage, pageSize, mapper);
		} catch (Exception e) {
			throw new IllegalArgumentException(e);
		}finally{
			server.release(searcher);
		}
		if(list != null){
			result.addAll(list);
		}
		return result;
	}
	
	public static void main(String[] args) {
		System.out.println(DateTime.parse("2012-12-12"));;
	}
}
